package xyz.xiaoshengfu.xrpc.consumer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.lang3.StringUtils;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import xyz.xiaoshengfu.xrpc.common.MarshallingCodeCFactory;
import xyz.xiaoshengfu.xrpc.config.ProviderServiceInfor;
import xyz.xiaoshengfu.xrpc.data.XRpcRequestData;
import xyz.xiaoshengfu.xrpc.data.XRpcResources;
import xyz.xiaoshengfu.xrpc.data.XRpcResponseData;
import xyz.xiaoshengfu.xrpc.data.XRpcResult;

/**
 * <p>Title: XRpcConsumerInvokeClient</p>
 * <p>Description: xRPC消费者执行客户端</p>
 * @author xiaoshengfu(2439323118@qq.com)
 * @date 2020年2月25日
 */
public final class XRpcConsumerInvokeClient {

	private static final Map<String, XRpcConsumerInvokeClient> invokeClientMap = new ConcurrentHashMap<String, XRpcConsumerInvokeClient>();
	private final ConcurrentHashMap<Long, XRpcResult> xRpcResultMap = new ConcurrentHashMap<Long, XRpcResult>();
	private final LinkedBlockingQueue<XRpcResult> unSendXRpcResultQueue = new LinkedBlockingQueue<XRpcResult>();

	private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);
	private Channel channel;
	private Bootstrap bootstrap;
	public volatile boolean auth = false;
	private ProviderServiceInfor providerServiceInfor;
	private int doConnectTimes = 0;
	private static AtomicLong requestALG = new AtomicLong(0);

	public XRpcConsumerInvokeClient(ProviderServiceInfor providerServiceInfor) {
		super();
		this.providerServiceInfor = providerServiceInfor;
	}

	public ConcurrentHashMap<Long, XRpcResult> getxRpcResultMap() {
		return xRpcResultMap;
	}

	public boolean isAuth() {
		return auth;
	}

	public void setAuth(boolean auth) {
		this.auth = auth;
	}

	public ProviderServiceInfor getProviderServiceInfor() {
		return providerServiceInfor;
	}

	public void cleanDoConnectTimes() {
		doConnectTimes = 0;
	}

	public static boolean applicationNameIsRegister(String applicationName) {
		return invokeClientMap.get(applicationName) == null ? false : true;
	}

	public static void registerXRpcConsumerInvokeClient(ProviderServiceInfor providerServiceInfor) {
		if (providerServiceInfor != null && StringUtils.isNotBlank(providerServiceInfor.getProviderAddress())
				&& StringUtils.isNotBlank(providerServiceInfor.getApplicationName())
				&& providerServiceInfor.getProviderPort() != null && providerServiceInfor.getServerState() != null) {
			invokeClientMap.remove(providerServiceInfor.getApplicationName());
			XRpcConsumerInvokeClient xRpcConsumerInvokeClient = new XRpcConsumerInvokeClient(providerServiceInfor);
			xRpcConsumerInvokeClient.start();
			invokeClientMap.put(providerServiceInfor.getApplicationName(), xRpcConsumerInvokeClient);
		}
	}

	public void sendResultQueue() {
		if (channel != null && channel.isActive()) {
			while (!unSendXRpcResultQueue.isEmpty()) {
				channel.writeAndFlush(unSendXRpcResultQueue.poll());
			}
		}
	}

	public void setReturnValueAndNotify(XRpcResponseData xRpcResponseData) {
		XRpcResult xRpcResult = xRpcResultMap.get(xRpcResponseData.getRequestID());
		if (xRpcResult != null) {
			synchronized (xRpcResult) {
				if (XRpcResponseData.OK == xRpcResponseData.getStatusCode()) {
					xRpcResult.setReturnValue(xRpcResponseData.getReturnValue());
					xRpcResult.setStatusCode(XRpcResult.OK);
				} else if (xRpcResponseData.getStatusCode() == XRpcResponseData.EXCEPTION) {
					xRpcResult.setException(xRpcResponseData.getException());
					xRpcResult.setStatusCode(XRpcResult.EXCEPTION);
				} else if (xRpcResponseData.getStatusCode() == XRpcResponseData.TIMEOUT) {
					xRpcResult.setStatusCode(XRpcResult.TIMEOUT);
				} else if (xRpcResponseData.getStatusCode() == XRpcResponseData.ERROR) {
					xRpcResult.setStatusCode(XRpcResult.ERROR);
				}
				xRpcResult.notify();
			}
			xRpcResultMap.remove(xRpcResponseData.getRequestID());
		}

	}

	public static XRpcResult invokeRemoteMethod(String fullyQualifiedClassName, String methodName, Object[] args) {
		XRpcRequestData xRpcRequestData = new XRpcRequestData();
		xRpcRequestData.setSendApplicationName(XRpcResources.getLocalApplicationName());
		xRpcRequestData.setSendApplicationName(XRpcResources.getLocalApplicationName());
		xRpcRequestData.setInterfaceName(fullyQualifiedClassName);
		xRpcRequestData.setRequestType(XRpcRequestData.RPC);
		xRpcRequestData.setMethodName(methodName);
		xRpcRequestData.setTimeoutTime(XRpcResources.getReferenceTimeout(fullyQualifiedClassName));
		if (args != null) {
			xRpcRequestData.setMethodParameters(args);
		} else {
			xRpcRequestData.setMethodParameters(new Object[0]);
		}
		xRpcRequestData.setRequestID(requestALG.addAndGet(1));
		String xRpcApplicationName = XRpcConsumerSubscribeClient.weightGetXRpcApplicationName(fullyQualifiedClassName);
		if (xRpcApplicationName != null) {
			xRpcRequestData.setAcceptApplicationName(xRpcApplicationName);
			XRpcConsumerInvokeClient xRpcConsumerInvokeClient = invokeClientMap.get(xRpcApplicationName);
			if (xRpcConsumerInvokeClient != null) {
				return xRpcConsumerInvokeClient.sendRequest(xRpcRequestData);
			}
		}
		return new XRpcResult(XRpcResult.NO_SERVICE_PROVIDER, fullyQualifiedClassName,
				xRpcRequestData.getTimeoutTime());
	}

	private XRpcResult sendRequest(XRpcRequestData xRpcRequestData) {
		XRpcResult xRpcResult = new XRpcResult(XRpcResult.INIT, xRpcRequestData.getInterfaceName(),
				XRpcResources.getReferenceTimeout(xRpcRequestData.getInterfaceName()));
		xRpcResult.setRequestID(xRpcRequestData.getRequestID());
		xRpcResult.setTimeoutTime(xRpcRequestData.getTimeoutTime());
		xRpcResult.setServiceApplicationName(xRpcRequestData.getAcceptApplicationName());
		xRpcResultMap.put(xRpcRequestData.getRequestID(), xRpcResult);
		if (channel != null && channel.isActive() && auth) {
			channel.writeAndFlush(xRpcRequestData);
		} else {
			unSendXRpcResultQueue.add(xRpcResult);
		}
		return xRpcResult;
	}

	public static void removeErrorClient(String applicationName) {
		invokeClientMap.remove(applicationName);
	}

	private void sendAuthRequest() {
		XRpcRequestData xRpcRequestData = new XRpcRequestData();
		xRpcRequestData.setSendApplicationName(XRpcResources.getLocalApplicationName());
		xRpcRequestData.setAcceptApplicationName(providerServiceInfor.getApplicationName());
		xRpcRequestData.setRequestType(XRpcRequestData.AUTH);
		if (channel != null && channel.isActive() && !auth) {
			channel.writeAndFlush(xRpcRequestData);
		}
	}

	public static void removeTimeOutXRpcResult(String applcatinName, Long requestID) {
		invokeClientMap.get(applcatinName).getxRpcResultMap().remove(requestID);
	}

	public void start() {
		try {
			bootstrap = new Bootstrap();
			bootstrap.group(workGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
				protected void initChannel(SocketChannel socketChannel) throws Exception {
					ChannelPipeline p = socketChannel.pipeline();
					p.addLast(new IdleStateHandler(5, 5, 5));
					p.addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
					p.addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
					p.addLast(new XRpcConsumerInvokeClientHandler(XRpcConsumerInvokeClient.this));
				}
			});
			doConnect();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void fixXRpcResultError() {
		for (XRpcResult xRpcResult : unSendXRpcResultQueue) {
			synchronized (xRpcResult) {
				xRpcResult.setStatusCode(XRpcResult.ERROR);
				xRpcResult.notify();
			}
		}
		for (XRpcResult xRpcResult : xRpcResultMap.values()) {
			synchronized (xRpcResult) {
				xRpcResult.setStatusCode(XRpcResult.ERROR);
				xRpcResult.notify();
			}
		}
	}

	/**
	  *<p>Title: doConnect</p>
	  *<p>Description: 与服务提供者连接</p>
	 */
	public void doConnect() {
		//System.out.println(providerServiceInfor);
		doConnectTimes++;
		if (channel != null && channel.isActive()) {
			doConnectTimes = 0;
			return;
		}
		if (doConnectTimes > 3) {
			providerServiceInfor.setServerState(ProviderServiceInfor.ERROR);
			invokeClientMap.remove(providerServiceInfor.getApplicationName());
			XRpcConsumerSubscribeClient.removeErrorApplicationServer(providerServiceInfor.getApplicationName());
			fixXRpcResultError();
			return;
		}
		providerServiceInfor.setServerState(ProviderServiceInfor.INIT);
		ChannelFuture future = bootstrap.connect(providerServiceInfor.getProviderAddress(),
				providerServiceInfor.getProviderPort());
		future.addListener(new ChannelFutureListener() {
			public void operationComplete(ChannelFuture futureListener) throws Exception {
				if (futureListener.isSuccess()) {
					channel = futureListener.channel();
					doConnectTimes = 0;
					sendAuthRequest();
					//System.out.println(XRpcResources.getLocalApplicationName() + ":" + providerServiceInfor.getApplicationName() + "连接成功！" + doConnectTimes);
				} else {
					futureListener.channel().eventLoop().schedule(new Runnable() {
						@Override
						public void run() {
							doConnect();
						}
					}, 5, TimeUnit.SECONDS);
					//System.out.println(XRpcResources.getLocalApplicationName() + ":" + providerServiceInfor.getApplicationName() + "连接失败！" + doConnectTimes);
				}
			}
		});
	}
}
